package com.citusdata.elven.feeder;

import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import com.citusdata.elven.linearRoad.LinearRoadService;
import com.google.protobuf.Message;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;


public class CarDataFeeder {
  private class Fields {
    private Fields() { }

    // Used as array indices, these do not fit into Java's Enum construct.
    public static final int TYPE = 0;
    public static final int TIME = 1;
    public static final int VID  = 2;
    public static final int SPEED = 3;
    public static final int XWAY  = 4;
    public static final int LANE  = 5;
    public static final int DIR   = 6;
    public static final int SEG   = 7;
    public static final int POS   = 8;
    public static final int QID   = 9;
    public static final int SINIT = 10;
    public static final int SEND  = 11;
    public static final int DOW   = 12;
    public static final int TOD   = 13;
    public static final int DAY   = 14;
  }

  private class ResponseType {
    private ResponseType() { }
    
    // Used only as integer values, these do not fit into Java's Enum construct.
    public static final int TOLL     = 0;
    public static final int ACCIDENT = 1;
    public static final int ACCOUNT  = 2;
    public static final int EXPENDITURE = 3;
  }


  public static final String SEPARATOR = ",";

  public static Message parseRequest(String line) throws IOException {
    Message request = null;
    String[] elements = line.split(SEPARATOR);
    
    switch(Integer.valueOf(elements[Fields.TYPE])) {
    case 0: {
      PositionRequest.Builder position = PositionRequest.newBuilder();
      position.setTime(Integer.valueOf(elements[Fields.TIME]));
      position.setVid(Integer.valueOf(elements[Fields.VID]));
      position.setSpeed(Integer.valueOf(elements[Fields.SPEED]));
      position.setXway(Integer.valueOf(elements[Fields.XWAY]));
      position.setLane(Integer.valueOf(elements[Fields.LANE]));
      position.setDir(Integer.valueOf(elements[Fields.DIR]));
      position.setSeg(Integer.valueOf(elements[Fields.SEG]));
      position.setPos(Integer.valueOf(elements[Fields.POS]));

      request = position.build();
      break;
    }
    case 2: {
      AccountRequest.Builder account = AccountRequest.newBuilder();
      account.setTime(Integer.valueOf(elements[Fields.TIME]));
      account.setVid(Integer.valueOf(elements[Fields.VID]));
      account.setQid(Integer.valueOf(elements[Fields.QID]));

      request = account.build();
      break;
    }
    case 3: {
      ExpenditureRequest.Builder expenditure = ExpenditureRequest.newBuilder();
      expenditure.setTime(Integer.valueOf(elements[Fields.TIME]));
      expenditure.setVid(Integer.valueOf(elements[Fields.VID]));
      expenditure.setXway(Integer.valueOf(elements[Fields.XWAY]));
      expenditure.setQid(Integer.valueOf(elements[Fields.QID]));
      expenditure.setDay(Integer.valueOf(elements[Fields.DAY]));

      request = expenditure.build();
      break;
    }
    default:
      break;    // Ignore others
    }

    return request;
  }


  public static int getRequestSemiMinute(String line) {
    String[] elements = line.split(SEPARATOR);
    int second = Integer.valueOf(elements[Fields.TIME]);

    // LinearRoadService can handle out-of-order requests properly as long as
    // all requests for the previous semi-minute are processed before new
    // requests for the current semi-minute are handled. This semi-minute
    // restriction follows from the following requirements:
    // (1) cars generate position reports every 30 seconds, and ordered reports
    // are necessary for accident detection, and
    // (2) speed averages for the previous minutes are calculated and then
    // reported to the client, and "late" position reports will result in
    // mis-calculated speed averages that are already reported to the client.

    int semiMinute = ((second/30) + 1);
    return semiMinute;
  }
  

  public static Message dispatchRequest(LinearRoadService linearRoad, 
                                        Message request) {
    Message response = null;

    if (linearRoad.localStorage()) {

      if (request instanceof PositionRequest) {
        response = linearRoad.reportPosition((PositionRequest) request);

        PositionResponse position = (PositionResponse) response;
        if (position.hasAssessment()) {
          TollAssessment toll = position.getAssessment();
          linearRoad.putHistoricalToll(toll.getVid(), toll.getTime(), toll);
        }
      } else if (request instanceof AccountRequest) {
        response = linearRoad.computeAccount((AccountRequest) request);
      } else if (request instanceof ExpenditureRequest) {
        response = linearRoad.computeExpenditure((ExpenditureRequest) request);
      }

    } else {
      // If no local storage, only execute computational requests
      if (request instanceof PositionRequest) {
        response = linearRoad.reportPosition((PositionRequest) request);
      }
    }

    return response;
  }


  public static String serializeResponse(Message response) {
    StringBuffer textResponse = new StringBuffer();

    if (response instanceof PositionResponse) {
      PositionResponse posResp = (PositionResponse) response;

      if (posResp.hasToll()) {
        TollNotification toll = posResp.getToll();
        textResponse.append(ResponseType.TOLL).append(SEPARATOR);
        textResponse.append(toll.getVid()).append(SEPARATOR);
        textResponse.append(toll.getTime()).append(SEPARATOR);
        textResponse.append(toll.getEmitTime()).append(SEPARATOR);
        textResponse.append(toll.getSegSpeed()).append(SEPARATOR);
        textResponse.append(toll.getToll());
      } 

      if (posResp.hasAccident()) {
        // Append accident information to toll notification
        textResponse.append(System.getProperty("line.separator"));

        AccidentNotification accident = posResp.getAccident();
        textResponse.append(ResponseType.ACCIDENT).append(SEPARATOR);
        textResponse.append(accident.getTime()).append(SEPARATOR);
        textResponse.append(accident.getEmitTime()).append(SEPARATOR);
        textResponse.append(accident.getVid()).append(SEPARATOR);
        textResponse.append(accident.getAccidSegment());
      }

    } else if (response instanceof AccountResponse) {
      AccountResponse account = (AccountResponse) response;
      textResponse.append(ResponseType.ACCOUNT).append(SEPARATOR);
      textResponse.append(account.getTime()).append(SEPARATOR);
      textResponse.append(account.getEmitTime()).append(SEPARATOR);
      textResponse.append(account.getQid()).append(SEPARATOR);
      textResponse.append(account.getResultTime()).append(SEPARATOR);
      textResponse.append(account.getBalance());

    } else if (response instanceof ExpenditureResponse) {
      ExpenditureResponse expenditure = (ExpenditureResponse) response;
      textResponse.append(ResponseType.EXPENDITURE).append(SEPARATOR);
      textResponse.append(expenditure.getTime()).append(SEPARATOR);
      textResponse.append(expenditure.getEmitTime()).append(SEPARATOR);
      textResponse.append(expenditure.getQid()).append(SEPARATOR);
      textResponse.append(expenditure.getBalance());

    }

    if (textResponse.length() == 0) {
      return null;
    } else {
      return textResponse.toString();
    }
  }


  public static void main(String[] args) throws IOException {
    if (args.length < 1) {
      System.err.println("Usage: CarDataFeeder type0_file [output_file]");
      System.exit(-1);
    }

    BufferedReader inputStream = null;
    PrintWriter    outputStream = null;
    try {
      String inputFileName  = args[0];
      String outputFileName = "/var/tmp/notifications.out";
      if (args.length == 2) {
        outputFileName = args[1];
      }
      inputStream  = new BufferedReader(new FileReader(inputFileName));
      outputStream = new PrintWriter(new FileWriter(outputFileName));

      LinearRoadService linearRoad = new LinearRoadService();
      String inputLine;

      while ((inputLine = inputStream.readLine()) != null) {
        Message request = parseRequest(inputLine);
        Message response = dispatchRequest(linearRoad, request);
        String textResponse = serializeResponse(response);
        if (textResponse != null) {
          outputStream.println(textResponse);
        }
      }
      
    } finally {
      if (inputStream != null) {
        inputStream.close();
      }
      if (outputStream != null) {
        outputStream.close();
      }
    }
  }
}